package org.kth.dks.dks_comm;

/**
 * <p>Title: DKS</p>
 * <p>Description: DKS Middleware</p>
 * <p>Copyright: Copyright (c) 2004</p>
 * <p>Company: KTH-IMIT/SICS</p>
 * @author Ali Ghodsi
 * @version 1.0
 */

import java.net.Socket;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;
import org.kth.dks.dks_exceptions.DKSRefNoResponse;
import org.kth.dks.dks_marshal.DKSMarshal;
import org.kth.dks.dks_marshal.DKSMessage;
import org.kth.dks.dks_marshal.MsgSrcDestWrapper;
import org.kth.dks.util.Pair;

public class ConnectionHandler {

  private Logger log = Logger.getLogger(ConnectionHandler.class);
  private Map handlerMap                = new HashMap();
  private int statMsgsDelivered   = 0;
  private int statMsgsFailed      = 0;
  private int statBytesReceived   = 0;
  private int statMsgsReceived    = 0;
  private int statBytesSent       = 0;
  private int statOpenConnections = 0;
  private int statTotalConnections= 0;
  private int statUnackedMsgs     = 0;
  private DKSMarshal marshal            = null;
  private ConnectionManager cm;

  static {
    DKSMessage.addMessageTypePrefixed("BOOT", "dks_marshal.BootstrapMsg");
  }

  public ConnectionHandler(DKSMarshal m, ConnectionManager c) {
    marshal = m;
    cm      = c;
  }

  public void statAddMsgsDelivered(int i) { statMsgsDelivered+=i; }
  public void statAddMsgsFailed(int i)    { statMsgsFailed+=i; }
  public void statAddBytesReceived(int i) { statBytesReceived+=i; }
  public void statAddMsgsReceived(int i)  { statMsgsReceived+=i; }
  public void statAddMsgsUnacked(int i)   { statUnackedMsgs+=i; }
  public void statAddBytesSent(int i)     { statBytesSent+=i; }
  public void statAddOpenConnection(int i){ statOpenConnections+=i; }
  public void statAddTotalConnection(int i){ statTotalConnections+=i; }
  public int getBytesRec()                { return statBytesReceived; }
  public int getMsgsRec()                 { return statMsgsReceived; }
  public int getBytesSent()               { return statBytesSent; }
  public int getMsgsDelivered()           { return statMsgsDelivered; }
  public int getMsgsFailed()              { return statMsgsFailed; }
  public int getNumConnections()          { return handlerMap.size(); }
  public int getOpenConnections()         { return statOpenConnections; }
  public int getTotalConnections()        { return statTotalConnections; }
  public int getMsgsUnackedNB()           { return statUnackedMsgs; }

  public int getMsgsUnacked() {
    int acc = 0;
    for (Iterator it = handlerMap.values().iterator(); it.hasNext();) {
      ConnHandlerOut cout = (ConnHandlerOut)((Pair)it.next()).second();
      acc += cout.getUnackedMsgs();
    }
    return acc;
  }

  public final DKSMarshal getDKSMarshal()       { return marshal; }

  public double getNodeRTT(DKSRef node) {
    Pair p = node==null ? null : (Pair) handlerMap.get(node.getDKSNetAddress());
    ConnHandlerOut cout = (p==null ? null : (ConnHandlerOut)p.second());

    if (cout==null) {
      return ConnHandlerOut.getDefaultRTO();
    }

    return cout.getRTO();
  }


  public synchronized void createConnection(Socket s, DKSNetAddress na) throws DKSRefNoResponse
  {
    if (handlerMap.containsKey(na)) {
      log.error( "createConnection: A connection already exists to "+na);
      return;
    }

    ConnHandlerOut cout;
    if (s!=null)
	cout = new ConnHandlerOut(this, na, s);
    else {
	DKSNetAddress myNA = new DKSNetAddress(cm.getHostAddress(), cm.getLocalPort());

	cout = new ConnHandlerOut(this, myNA, na);
	s = cout.getSocket();
    }
    ConnHandlerIn  cin  = new ConnHandlerIn(this, s, na, cout);
    //    cin.setConnOut(cout);
    // cout.setConnIn(cin);
    handlerMap.put(na, new Pair(cin, cout));
  }

  public void removeConnection(DKSNetAddress n) {
    synchronized (handlerMap) {
      if (handlerMap.containsKey(n)) {
    	  Pair p = (Pair) handlerMap.get(n);
    	  ConnHandlerIn cin = (ConnHandlerIn) p.first();
    	  ConnHandlerOut cout = (ConnHandlerOut) p.second();
    	  cin.end();
    	  cout.end();
    	  handlerMap.remove(n);
      }
    }
  }
  
  public synchronized Set getAllPeers() {
	  return new HashSet(handlerMap.keySet());
  }

  public synchronized boolean send(MsgSrcDestWrapper triple)
  {

    DKSNetAddress src = triple.getSrc().getDKSNetAddress();
    DKSNetAddress dest = triple.getDest().getDKSNetAddress();

    if (!handlerMap.containsKey(dest)){
      try {
        createConnection(null, dest);
      } catch (DKSRefNoResponse ex) {
        return false;
      }
    }
    Pair p = (Pair) handlerMap.get(dest);
    ConnHandlerOut o = (ConnHandlerOut) p.second();

    o.sendMessage(src, triple);
    return true;
  }

  public void end() {
    synchronized (handlerMap) {
      for (Iterator it = handlerMap.values().iterator(); it.hasNext(); ) {
        Pair p = (Pair) it.next();
        ConnHandlerIn cin = (ConnHandlerIn) p.first();
        ConnHandlerOut cout = (ConnHandlerOut) p.second();
        cin.end();
        cout.end();
      }
    }
  }

}
